package com.atguigu.hotitems_analysis

import java.lang
import java.sql.Timestamp
import java.util.Properties

import kafka.controller.KafkaController
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.runtime.state.memory.MemoryStateBackend
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer


/**
 * 统计商品TopN热门排行，1小时窗口5分钟滑动
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.hotitems_analysis
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/10 18:23
 */
object HotItems {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironment(1)

    //Flink的并行度设置要和Kafka的分区数一致
//    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE)
    //设置启动检查点时间
    env.enableCheckpointing(1000L)
    //检查点必须在一分钟内完成，否则将被丢弃
    env.getCheckpointConfig.setCheckpointTimeout(6000L)
    // 同时只允许一个检查点处于进程中
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1);
    env.setNumberOfExecutionRetries(2)
//    env.setStateBackend(new MemoryStateBackend())
    val backend = new RocksDBStateBackend("/StateBackend")
    env.setStateBackend(backend)

    val path: String = HotItems.getClass.getClassLoader.getResource("UserBehavior.csv").getPath

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "hadoop105:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    //偏移量的自动重置(最近的offset)
    properties.setProperty("auto.offset.reset", "latest")


    val value = env.readTextFile(path)
//    val value = env.addSource(new FlinkKafkaConsumer[String]("hotitmes", new SimpleStringSchema, properties))
      .map(
        data => {
          val dataString: Array[String] = data.split(",")
          UserBehavior(dataString(0).trim.toLong,
            dataString(1).trim.toLong,
            dataString(2).trim.toInt,
            dataString(3).trim,
            dataString(4).trim.toLong)
        }
      )
      .filter(_.behavior == "pv")
      .assignAscendingTimestamps(_.timestamp * 1000L) //开启watermark
      .keyBy(_.itemId) //根据商品id进行分组
      .timeWindow(Time.hours(1), Time.minutes(5)) //按照一小时开窗，5分钟滑动窗口
      .aggregate(new MyAggregate(), new MyWindowFunction()) //统计每个窗口中的商品个个点击数
      .keyBy(_.windowEnd) //根据windowEnd做最后的汇总
      .process(new MyProcess(3))

    value.print()

    env.execute("HotItems TopN")
  }

}

//样例类
case class UserBehavior(
                         userId: Long,
                         itemId: Long,
                         categoryId: Int,
                         behavior: String,
                         timestamp: Long
                       )

case class ItemViewCount(
                          itemId: Long,
                          windowEnd: Long,
                          count: Long
                        )

//自定义简单的聚合函数
class MyAggregate() extends AggregateFunction[UserBehavior, Long, Long] {
  override def createAccumulator(): Long = 0L

  override def add(in: UserBehavior, acc: Long): Long = acc + 1

  override def getResult(acc: Long): Long = acc

  override def merge(acc: Long, acc1: Long): Long = acc + acc1
}

/**
 * 第一个lang是MyAggregate输出的类型
 */
class MyWindowFunction() extends WindowFunction[Long, ItemViewCount, Long, TimeWindow] {

  override def apply(key: Long, window: TimeWindow, input: Iterable[Long], out: Collector[ItemViewCount]): Unit = {
    //MyAggregate聚合的结果只有一个
    out.collect(ItemViewCount(key, window.getEnd, input.last))
  }
}

//自定义处理
class MyProcess(topN: Int) extends KeyedProcessFunction[Long, ItemViewCount, String] {

  var listState: ListState[ItemViewCount] = _

  override def open(parameters: Configuration): Unit = {
    listState = getRuntimeContext.getListState(new ListStateDescriptor[ItemViewCount]("listState", classOf[ItemViewCount]))
  }

  override def processElement(i: ItemViewCount, context: KeyedProcessFunction[Long, ItemViewCount, String]#Context, collector: Collector[String]): Unit = {
    listState.add(i)

    //设置定时器
    context.timerService().registerEventTimeTimer(i.windowEnd + 1)
  }

  //定时器触发器
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, ItemViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
    //取出来的是java的Iterable迭代器
    val viewCounts: lang.Iterable[ItemViewCount] = listState.get()

    val list = new ListBuffer[ItemViewCount]
    import scala.collection.JavaConversions._
    for (elem <- viewCounts) {
      list += elem
    }

    //排序输出前N
    val counts: ListBuffer[ItemViewCount] = list.sortWith((A, B) => A.count > B.count).take(topN)


    val sb = new StringBuilder()

    sb.append("==================================\n")
    sb.append("时间：").append(new Timestamp(timestamp - 1)).append("\n")

    for (elem <- counts.indices) {
      sb.append("NO:").append(elem + 1)
        .append(" 商品ID:").append(counts(elem).itemId)
        .append(" 浏览量：").append(counts(elem).count).append("\n")
    }

    //清空状态
    listState.clear()

    Thread.sleep(500L)

    out.collect(sb.toString())
  }
}